package com.wordcount_v2;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.net.URI;

/**
 * @author Administrator
 */
public class WordCountJobV2 extends Configured implements Tool {

    public static final String HDFS_URL = "hdfs://aliyun:8020";
    private static final String HADOOP_USER_NAME = "root";

    public static void main(String[] args) throws Exception {
        //执行job作业任务类的对象是谁
        ToolRunner.run(new WordCountJobV2(), args);
    }

    @Override
    public int run(String[] args) throws Exception {
        int argsNum = 2;
        // 文件输入路径和输出路径由外部传参指定
        if (args.length < argsNum) {
            System.out.println("Input and output paths are necessary!");
            return 0;
        }
        System.setProperty("HADOOP_USER_NAME", HADOOP_USER_NAME);
        Configuration configuration = new Configuration();
        configuration.set("fs.defaultFS", HDFS_URL);

        // 创建job
        Job job = Job.getInstance(configuration);
        job.setJarByClass(WordCountJobV2.class);

        // 设置 map
        job.setMapperClass(WordCountMapperV2.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        // 设置 Combiner
        job.setCombinerClass(WordCountReducerV2.class);

        // 设置 reduce
        job.setReducerClass(WordCountReducerV2.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        job.setNumReduceTasks(WordCountDataUtils.WORD_LIST.size());

        // 设置 inputFormat 和 outputFormat
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job, new Path(args[0]));

        // 设置自定义分区
        job.setPartitionerClass(CustomPartitioner.class);

        // 如果输出目录已经存在，则必须先删除，否则重复运行程序时会抛出异常
        FileSystem fileSystem = FileSystem.get(new URI(HDFS_URL), configuration, HADOOP_USER_NAME);
        Path outputPath = new Path(args[1]);
        if (fileSystem.exists(outputPath)) {
            fileSystem.delete(outputPath, true);
        }
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job, outputPath);

        // 将作业提交到群集并等待它完成，参数设置为 true 代表打印显示对应的进度
        boolean completion = job.waitForCompletion(true);

        // 关闭之前创建的 fileSystem
        fileSystem.close();

        // 根据作业结果,终止当前运行的 Java 虚拟机,退出程序
        System.exit(completion ? 0 : -1);
        return completion ? 0 : -1;
    }
}
